#include "app_pool.h"
#include <mqueue.h>
#include <unistd.h>
#include <pthread.h>
#include "log/log.h"
#include <stdlib.h>
#include <fcntl.h>

// 消息队列的路径
static const char *mq_name = "/app_pool_mq";
static mqd_t mq;  // 消息队列描述符
//创建线程池
static pthread_t *threads;
static int g_thread_count=0;  //县城池的线程数量 

//线程函数
static void *thread_func(void *argv){
    Task task;
    while(1){
        int len = mq_receive(mq,(char*)&task,sizeof(Task),NULL);
        if(len == sizeof(Task)){
            if(task.task_fun){
                task.task_fun(task.arg);
            }
        }else{
            log_error("mq_receive error");
        }
    }
    return NULL; // 不会执行到
}
// 初始化线程池
int app_pool_init(int thread_count)
{
    // 创建打开消息队列
    struct mq_attr attr;
    //消息队列中最大消息数量
    attr.mq_maxmsg = 10;
    //消息队列中每条消息的最大长度
    attr.mq_msgsize = sizeof(Task);
    // 打开消息队列
    mq = mq_open(mq_name, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == -1)
    {
        log_error("mq_open error");
        return -1;
    }
   //启动多个线程并用一个容器存储起来
    g_thread_count=thread_count;
    //申请内存空间
    threads=malloc(sizeof(pthread_t)*g_thread_count);
    for (int i = 0; i < thread_count; i++){
        pthread_create(&threads[i], NULL, thread_func, NULL);
    }
    return 0;
}
// 销毁线程池
void app_pool_destroy(void){
    //关闭并删除消息队列
    mq_close(mq);
    mq_unlink(mq_name);
    //销毁线程池
    for(int i=0;i<g_thread_count;i++){
        pthread_cancel(threads[i]);
        pthread_join(threads[i],NULL);
    }
    free(threads);
}
// 添加任务到线程池
int app_pool_addTask(int (*task_fun)(void *arg), void *arg){
    //创建任务
    Task task={
        .task_fun=task_fun,
        .arg=arg,
    };

    //发送消息到消息队列
    if(mq_send(mq,(const char*)&task,sizeof(Task),0)==-1){
        log_error("mq_send error");
        return -1;
    }
    return 0;
}